package ex.tableapi;

import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.*;


/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/tableapi/#groupby-window-aggregation
 *
 * 更换jar ：  https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc/3.1.1-1.17
 */
public class Ex05 extends ApiFrame{
    private String outName = "printOutTable";
    private    String createPrintOutDDL = "CREATE TABLE printOutTable (" +
            "a STRING, " +
            "c1  BIGINT, " +
            "c2  BIGINT, " +
            "c3  BIGINT " +
            ") WITH (" +
            " 'connector' = 'print' " +
            ")";

    public static void main(String[] args) {
        Ex05 demo =new Ex05();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        demo.registerConsole(demo.createPrintOutDDL,demo.outName);


        Table result = orders
                // define window
                .window(
                        Over
                                .partitionBy($("a"))
                                .orderBy($("rowtime"))
                                .preceding(UNBOUNDED_RANGE)
                                .following(CURRENT_RANGE)
                                .as("w"))
                // sliding aggregate
                .select(
                        $("a"),
                        $("b").avg().over($("w")).as("c1"),
                        $("b").max().over($("w")).as("c2"),
                        $("b").min().over($("w")).as("c3")
                );

        result.printSchema();

        result.executeInsert(demo.outName);
        // print
//        result.execute().print();

    }
}
